package com.zr.domain.bar.service.fun;

import com.zr.domain.bar.entity.MarketDataEvent;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public  class KLineCalculationProcessFunction extends ProcessWindowFunction<MarketDataEvent, MarketDataEvent, String, TimeWindow> {
    @Override
    public void process(String key, Context context, Iterable<MarketDataEvent> elements, Collector<MarketDataEvent> out) {
        MarketDataEvent firstKLine = null;
        double high = Double.MIN_VALUE;
        double low = Double.MAX_VALUE;

        for (MarketDataEvent marketData : elements) {
            if (firstKLine == null) {
                firstKLine = new MarketDataEvent(marketData.getChannel(),marketData.getInstrument(),marketData.getContract(), marketData.getTimestamp(), marketData.getOpen(), marketData.getHigh(), marketData.getLow(), marketData.getClose());
            } else {
                high = Math.max(high, marketData.getHigh());
                low = Math.min(low, marketData.getLow());
            }
        }

        if (firstKLine != null) {
            firstKLine.setHigh(high);
            firstKLine.setLow(low);
            out.collect(firstKLine);
        }
    }
}
